package com.udf.flink.scala.apitest.join

  import java.lang

  import org.apache.flink.api.common.functions.{CoGroupFunction, JoinFunction}
  import org.apache.flink.api.scala._
  import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
  import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  import org.apache.flink.streaming.api.scala.function.RichWindowFunction
  import org.apache.flink.streaming.api.watermark.Watermark
  import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
  import org.apache.flink.streaming.api.windowing.time.Time
  import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  import org.apache.flink.util.Collector
  import java.text.SimpleDateFormat
  import java.time.Duration
  import java.util.Date

  import com.alibaba.fastjson.JSON
  import com.alibaba.fastjson.serializer.SerializerFeature
  import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}

  import scala.beans.BeanProperty

object Const {
  final val names = List("wang", "li", "zhao", "qian", "sun")
  final val ages = List(0, 10, 20, 30)
  final val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
}
case class People(
                   @BeanProperty var id: Int = 0,
                   @BeanProperty var name: String = "",
                   @BeanProperty var age: Int = 0,
                   @BeanProperty var eventTime: Long = 0,
                   @BeanProperty var eventTimeStr: String = "",
                   @BeanProperty var joined: Long = 0,
                 ) {

  override def toString: String = {
    JSON.toJSONString(this, SerializerFeature.QuoteFieldNames)
  }
}

object People {
  def newPeopleByName(id: Int, name: String) = {
    val than = new People()
    val now = new Date()
    than.id = id
    than.name = name
    than.eventTime = now.getTime
    than.eventTimeStr = Const.dateFormat.format(now)
    than
  }

  def newPeopleByAge(id: Int, age: Int) = {
    val than = new People()
    val now = new Date()
    than.id = id
    than.age = age
    than.eventTime = now.getTime
    than.eventTimeStr = Const.dateFormat.format(now)
    than
  }
}




object  LeftJoinCoGroup {
    def main(args: Array[String]): Unit = {
      val env = StreamExecutionEnvironment.getExecutionEnvironment
      env.setParallelism(1)
      val WIN = 60 // s

      // id,name
      val nameStream = env
        //      .addSource(new NameSource)
        .socketTextStream("localhost", 9999)
        .filter(_.nonEmpty)
        .map(x => {
          val arr = x.split(",")
          People.newPeopleByName(id = arr(0).toInt, name = arr(1))
        }).assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(20))
        .withTimestampAssigner(new SerializableTimestampAssigner[People]{
          override def extractTimestamp(element: People, recordTimestamp: Long): Long = element.eventTime
        })
      )
//        .assignTimestampsAndWatermarks(
//          new BoundedOutOfOrdernessTimestampExtractor[People](Time.seconds(10L)) {
//            override def extractTimestamp(element: People): Long = element.eventTime
//          })

      // id,age
      val ageStream = env
        //      .addSource(new AgeSource)
        .socketTextStream("localhost", 9000)
        .filter(_.nonEmpty)
        .map(x => {
          val arr = x.split(",")
          People.newPeopleByAge(id = arr(0).toInt, age = arr(1).toInt)
        }).assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(20))
          .withTimestampAssigner(new SerializableTimestampAssigner[People]{
            override def extractTimestamp(element: People, recordTimestamp: Long): Long = element.eventTime
          })
      )
//        .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[People] {
//          val maxOutOfOrderness = 1000L // 最大无序数据到达的时间，用来生成水印,10s
//          var currentMaxTimestamp: Long = _
//
//          override def getCurrentWatermark: Watermark = {
//            new Watermark(currentMaxTimestamp - maxOutOfOrderness)
//          }
//
//          override def extractTimestamp(element: People, previousElementTimestamp: Long): Long = {
//            currentMaxTimestamp = Math.max(element.eventTime, currentMaxTimestamp)
//            element.eventTime
//          }
//        })

      nameStream
        .map(("name:", _))
        .print()

      ageStream
        .map(("age:", _))
        .print()

      // coGroup
      nameStream
        .coGroup(ageStream)
        .where(_.id).equalTo(_.id)
        .window(TumblingEventTimeWindows.of(Time.seconds(WIN)))
        .apply(new CoGroupFunction[People, People, People] {
          override def coGroup(first: lang.Iterable[People], second: lang.Iterable[People], out: Collector[People]): Unit = {
            if (first.iterator().hasNext) {
              first.forEach(left => {
                var noElement = true
                if (second.iterator().hasNext) {
                  second.forEach(rigth => {
                    noElement = false
                    val data = left.copy()
                    data.age = rigth.age
                    if (rigth.age > 0) {
                    data.joined = 1
                   }
//                    println(rigth.age,data.toString)
                    out.collect(data)
                  })
                }
                if (noElement) out.collect(left)
              })
            }

          }
        })
        .map(("-------- coGroup left Join", _))
        .print()

      // -------------------------------------------
      //   left join
      val nameStream2 = nameStream.map((_, "nameStream"))
      val ageStream2 = ageStream.map((_, "ageStream"))
      //    nameStream2
      //      .print()
      //
      //    ageStream2
      //      .print()

      nameStream2
        .join(nameStream2.union(ageStream2))
        .where(_._1.id)
        .equalTo(_._1.id)
        .window(TumblingEventTimeWindows.of(Time.seconds(WIN)))
        .apply(new JoinFunction[Tuple2[People, String], Tuple2[People, String], Tuple2[People, Boolean]] {
          override def join(first: (People, String), second: (People, String)): (People, Boolean) = {
            val data = first._1.copy()
            var same = true // 是否和自身连接
            if (second._2.equals("ageStream")) {
              same = false
              data.age = second._1.age
            }
            (data, same)
          }
        })
        .keyBy(_._1.id)
        .window(TumblingEventTimeWindows.of(Time.seconds(WIN)))
        .apply(new RichWindowFunction[Tuple2[People, Boolean], People, Int, TimeWindow] {
          override def apply(key: Int, window: TimeWindow, input: Iterable[(People, Boolean)], out: Collector[People]): Unit = {
            var len = 0
            var same: People = null
            var unSame: People = null
            for (i <- input.iterator) {
              len += 1
              if (i._2) same = i._1.copy() // a1,a1
              else unSame = i._1.copy() //  a1,a2
            }

            if (len % 2 == 0) out.collect(unSame) // 偶数返回 a1,a2
            else out.collect(same) // 奇数返回 a1,a1
          }
        })
        .map(("-------- left Join ", _))
        .print()


      env.execute("DataStreamCoGroup")
    }
  }